package service

import (
	"context"
	"encoding/json"
	"gitee.com/crack007/goose"
	"github.com/Shopify/sarama"
	"github.com/crack007/go-im/src/common/config"
	"github.com/crack007/go-im/src/common/constant"
	"github.com/crack007/go-im/src/common/dao"
	"github.com/crack007/go-im/src/common/entity"
	"github.com/crack007/go-im/src/common/model"
	"github.com/crack007/go-im/src/common/service"
	"sync"
)

type consumerService struct {
	queue    *service.KafkaConsumer
	lock     sync.Mutex
	msgQueue chan *model.MessageModel // 离线消息队列
}

func (s *consumerService) Bootstrap(ctx context.Context) error {
	go s.monitorNode(ctx)
	go s.offlineMessageQueue()
	return s.queueConsumer(ctx)
}

func (s *consumerService) queueConsumer(ctx context.Context) error {
	s.queue = service.NewKafkaConsumer(ctx)
	err := s.queue.Init()
	if err != nil {
		return err
	}
	return s.queue.Consumer(config.KafkaConfig.Topic, func(msg *sarama.ConsumerMessage) {
		var err error
		defer func() {
			if err != nil {
				goose.GetLogger().Error("Consumer msg error-> %s", err.Error())
			}
		}()
		messageModel := model.NewMessageModel()
		err = json.Unmarshal(msg.Value, messageModel)
		if err != nil {
			return
		}
		// 根据目标用户id，寻找对应的ws节点，并且重新投入消息队列，如果目标节点已下线，消息做离线处理
		var nodeModel *model.NodeModel
		// 寻找用户登录节点
		nodeModel = service.UserService.GetUser(messageModel.ToUser)
		s.msgQueue <- messageModel
		if nodeModel == nil {
			return
		}
		// 消息投放目标ws节点
		err = service.GrpcService.SendMsg(nodeModel, messageModel)
	})
}

func (s *consumerService) offlineMessageQueue() {
	if s.msgQueue == nil {
		panic("msgQueue为空")
	}
	for messageModel := range s.msgQueue {
		goose.GetLogger().Debug("%+v", messageModel)
		if messageModel.Action == constant.ACTION_CHAT {
			messageEntity := entity.NewMessageEntity()
			messageEntity.Message = messageModel.Data
			messageEntity.FromUID = messageModel.FromUser
			messageEntity.ToUID = messageModel.ToUser
			messageEntity.MsgType = messageModel.MessageType.Val()
			messageEntity.ReadStatus = 0
			dao.MessageDao.Save(messageEntity)
		}
		if messageModel.Action == constant.ACTION_FRIENDAPPLYRESULT {
			// 通知给发起的人
			resultModel := model.NewFriendApplyResultModel()
			json.Unmarshal([]byte(messageModel.Data), resultModel)
			noticeModel := model.NewNoticeModel()
			noticeModel.UserId = messageModel.ToUser
			noticeModel.FromUserId = messageModel.FromUser
			noticeModel.Title = "添加好友结果"
			noticeModel.Content = resultModel.Message
			service.NoticeService.PushNotice(noticeModel)
		}
		if messageModel.Action == constant.ACTION_FRIEND_APPLY {
			// 通知给接收的人
			noticeModel := model.NewNoticeModel()
			noticeModel.UserId = messageModel.ToUser
			noticeModel.FromUserId = messageModel.FromUser
			noticeModel.Title = "添加好友申请"
			noticeModel.Content = "您有新的好友申请，请及时处理"
			service.NoticeService.PushNotice(noticeModel)
		}
	}
	goose.GetLogger().Debug("consumerService.offlineMessageQueue quit")
}

func (s *consumerService) monitorNode(ctx context.Context) {
	etcdService := service.EtcdService
	etcdService.Init(ctx)
	etcdService.ListNode(constant.NodePath, func(etcdModel *model.EtcdModel) {
		service.NodeService.Add(etcdModel.Node)
	})
	etcdService.Watch(constant.NodePath, func(nodeName string, event string, etcdModel *model.EtcdModel) {
		if event == "DELETE" {
			goose.GetLogger().Info("Offline-> %s", nodeName)
			service.NodeService.Remove(nodeName)
		}
		if event == "PUT" {
			goose.GetLogger().Info("Online-> %s", etcdModel.Node.Id)
			service.NodeService.Add(etcdModel.Node)
		}
	})
}
